package com.homedo.distributed.id.client.gen.buffer;

import com.homedo.distributed.id.client.gen.constants.GenExecutorConstants;
import com.netflix.config.DynamicIntProperty;
import com.netflix.config.DynamicLongProperty;
import com.netflix.config.DynamicPropertyFactory;
import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.functions.Consumer;
import io.reactivex.schedulers.Schedulers;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;

public class GenMonitor {
    private static final Logger _LOGGER = LoggerFactory.getLogger(GenMonitor.class);

    private static DynamicLongProperty targetUsageSeconds = DynamicPropertyFactory.getInstance().getLongProperty("gen.monitor.targetUsageSeconds", TimeUnit.MINUTES.toSeconds(6));
    private static DynamicIntProperty multiple = DynamicPropertyFactory.getInstance().getIntProperty("gen.monitor.multiple", 1);
    private static DynamicIntProperty maxCacheSize = DynamicPropertyFactory.getInstance().getIntProperty("gen.monitor.maxCacheSize", 20000);
    private static DynamicIntProperty minCacheSize = DynamicPropertyFactory.getInstance().getIntProperty("gen.monitor.minCacheSize", 500);

    private ObservableEmitter<Integer> curEmitter;

    private ExecutorService monitorBufferExecutor = GenExecutorConstants.MONITOR_BUFFER_EXECUTOR;

    private GenBuffer genBuffer;
    private AtomicInteger monitorCount;
    private AtomicBoolean triggerForceFetch;

    public GenMonitor(GenBuffer genBuffer) {
        this.genBuffer = genBuffer;
        this.monitorCount = new AtomicInteger(0);
        this.triggerForceFetch = new AtomicBoolean(false);
    }


    private Observable<Integer> createObserver() {
        Observable.interval(1 * multiple.get(), 1 * multiple.get(), TimeUnit.SECONDS, Schedulers.from(monitorBufferExecutor)).subscribe(new Consumer<Long>() {
            @Override
            public void accept(Long aLong) throws Exception {
                if (!curEmitter.isDisposed()) {
                    curEmitter.onNext(monitorCount.get());// 延迟 1 * MULTIPLE 秒后 每隔1 * MULTIPLE秒发送一下统计数据
                    monitorCount.set(0);//重置统计数据
                }
            }
        });


        return Observable.create(
                new ObservableOnSubscribe<Integer>() {
                    @Override
                    public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                        curEmitter = emitter;
                    }
                }
        ).subscribeOn(Schedulers.from(monitorBufferExecutor));

    }

    public void start() {
        Observable<Integer> observable = this.createObserver();

        observable.buffer(6).subscribe(new Consumer<List<Integer>>() {
            @Override
            public void accept(List<Integer> counts) throws Exception {
                _LOGGER.debug("{}#{}使用统计:{}", genBuffer.getBizTag(), genBuffer.getGenType(), counts);

//                IntSummaryStatistics stats = counts.stream()
//                        .mapToInt((x) -> x)
//                        .summaryStatistics();
                double avgPerSeconds = counts.stream().mapToInt((x) -> x).average().getAsDouble() / multiple.get();

                double targetUsageSize = avgPerSeconds * targetUsageSeconds.get();//满足目标需要缓存个数

                int curCacheSize = genBuffer.getCacheSize();//当前缓存个数

                boolean satisfy = targetUsageSize < curCacheSize;
                if (!satisfy) {//不满足使用时长就扩容
                    int toCacheSize = ((int) targetUsageSize) + 1;
                    toCacheSize = toCacheSize > maxCacheSize.get() ? maxCacheSize.get() : toCacheSize;

                    if (toCacheSize > curCacheSize) {//前提是不大于最大缓存值
                        genBuffer.setCacheSize(toCacheSize);
                        _LOGGER.info("[扩容{},每个Buffer需要满足使用{}秒,每秒平均消费={},当前BufferSize={},需扩容至BufferSize={}]",
                                genBuffer.getBizTag() + "#" + genBuffer.getGenType(), targetUsageSeconds.get(), avgPerSeconds, curCacheSize, toCacheSize);


                        if (triggerForceFetch.compareAndSet(false, true)) {//动态扩容时实时触发
                            try {
                                _LOGGER.info("[扩容中start...]");
                                genBuffer.setForceFetch(true);
                                genBuffer.fetchGensAndDeal(genBuffer.getNextActivity(), true);
                                _LOGGER.info("[扩容end]");
                            } finally {
                                genBuffer.setForceFetch(false);
                                triggerForceFetch.compareAndSet(true, false);
                            }

                        }
                    }
                } else {
                    if (counts.stream().filter(e -> e > 0).count() < 3) {//半数不为0的时候再缩容
                        return;
                    }

                    //实际cacheSize 超过目标cacheSize+cacheSize*0.5就缩减 ；前提是不小于最低缓存值
                    int toCacheSize = (int) (targetUsageSize + targetUsageSize * 0.5);
                    toCacheSize = toCacheSize < minCacheSize.get() ? minCacheSize.get() : toCacheSize;

                    boolean needReduce = curCacheSize > toCacheSize;
                    if (needReduce) {
                        genBuffer.setCacheSize(toCacheSize);
                        _LOGGER.info("[缩容{},每个Buffer需要满足使用{}秒,每秒平均消费={},当前BufferSize={},需缩容至BufferSize={}]",
                                genBuffer.getBizTag() + "#" + genBuffer.getGenType(), targetUsageSeconds.get(), avgPerSeconds, curCacheSize, toCacheSize);
                    }
//                    System.out.println("满足:" + targetUsageSeconds.get() + " " + avgPerSeconds + " " + targetUsageSize + " " + curCacheSize);
                }

            }
        });
    }

    public void plusMonitorCount() {
        monitorCount.incrementAndGet();
    }
}
